package store

import (
	"encoding/json"
	"fmt"
	"github.com/hashicorp/raft-boltdb/v2"
	"io"
	"log"
	"net"
	"os"
	"path/filepath"
	"raftkv/utils"
	"sync"
	"time"
)
import "github.com/hashicorp/raft"

type ConsistencyLevel int

const (
	Default ConsistencyLevel = iota
	Stale
	Consistent
)

//常量
const (
	retainSnapshotCount = 2
	raftTimeout         = 10 * time.Second
	leaderWaitDelay     = 100 * time.Millisecond
	appliedWaitDelay    = 100 * time.Millisecond
)

type Store struct {
	lock     sync.Mutex
	db       map[string]string
	raft     *raft.Raft
	logger   *log.Logger
	RaftDir  string
	RaftBind string
}

// 指令操作
type command struct {
	Op    string `json:"op,omitempty"`
	Key   string `json:"key,omitempty"`
	Value string `json:"value,omitempty"`
}

func NewStore() *Store {
	return &Store{
		db:     make(map[string]string),
		logger: log.New(os.Stderr, "[store] ", log.LstdFlags),
	}
}

// canConsistentRead 是否能一致性读，既判断自己是否是领导节点
func (store *Store) canConsistentRead() error {
	err := store.raft.VerifyLeader().Error()
	if err != nil {
		return err
	}
	return nil
}

// Get 获取
func (store *Store) Get(key string, level ConsistencyLevel) (string, error) {
	if level != Stale {
		if store.raft.State() != raft.Leader {
			return "", raft.ErrNotLeader
		}
	}
	if level == Consistent {
		if err := store.canConsistentRead(); err != nil {
			//TODO：请求转发给主节点
			return "", err
		}
	}
	store.lock.Lock()
	defer store.lock.Unlock()
	return store.db[key], nil
}
func (store *Store) Set(key, value string) error {
	if store.raft.State() != raft.Leader {
		//todo: 转发给主节点
		return raft.ErrNotLeader
	}
	cmd := &command{
		Op:    "set",
		Key:   key,
		Value: value,
	}
	// JSON序列化，TODO:改进成proto buffer
	jsonData, err := json.Marshal(cmd)
	if err != nil {
		return err
	}
	//将日志数据应用与状态机
	future := store.raft.Apply(jsonData, raftTimeout)
	return future.Error()
}
// Delete .
func (s *Store) Delete(key string) error {
	if s.raft.State() != raft.Leader {
		return raft.ErrNotLeader
	}

	c := &command{
		Op:  "delete",
		Key: key,
	}
	b, err := json.Marshal(c)
	if err != nil {
		return err
	}

	f := s.raft.Apply(b, raftTimeout)
	return f.Error()
}

func (store *Store) SetMeta(key, value string) error {
	return store.Set(key, value)
}

// GetMeta 获取节点信息采用Stale模式
func (store *Store) GetMeta(key string) (string, error) {
	return store.Get(key, Stale)
}
// Join 加入集群
func (store *Store) Join(nodeID,httpAddr string,addr string)error{
	store.logger.Printf("received join request for remote node %s at %s", nodeID, addr)

	configFuture := store.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		store.logger.Printf("failed to get raft configuration: %v", err)
		return err
	}

	for _, srv := range configFuture.Configuration().Servers {
		// If a node already exists with either the joining node's ID or address,
		// that node may need to be removed from the config first.
		if srv.ID == raft.ServerID(nodeID) || srv.Address == raft.ServerAddress(addr) {
			// However if *both* the ID and the address are the same, then nothing -- not even
			// a join operation -- is needed.
			if srv.Address == raft.ServerAddress(addr) && srv.ID == raft.ServerID(nodeID) {
				store.logger.Printf("node %s at %s already member of cluster, ignoring join request", nodeID, addr)
				return nil
			}

			future := store.raft.RemoveServer(srv.ID, 0, 0)
			if err := future.Error(); err != nil {
				return fmt.Errorf("error removing existing node %s at %s: %s", nodeID, addr, err)
			}
		}
	}

	f := store.raft.AddVoter(raft.ServerID(nodeID), raft.ServerAddress(addr), 0, 0)
	if f.Error() != nil {
		return f.Error()
	}

	// Set meta info
	if err := store.SetMeta(nodeID, httpAddr); err != nil {
		return err
	}

	store.logger.Printf("node %s at %s joined successfully", nodeID, addr)
	return nil
}

// LeaderWithID 返回领导地址信息和id
func (store *Store) LeaderWithID() (string, string) {
	serverAddress, id := store.raft.LeaderWithID()
	fmt.Printf("领导ip：%v,ID：%v\n", serverAddress, id)
	return string(serverAddress), string(id)
}
func (store *Store) LeaderAPIAddr() string {
	_, id := store.LeaderWithID()
	if id == "" {
		return ""
	}
	addr, err := store.GetMeta(id)
	if err != nil {
		return ""
	}
	return addr
}

// Open raft启动类
func (store *Store) Open(enableSingle bool, localId string) error {
	config := raft.DefaultConfig()
	config.LocalID = raft.ServerID(localId)

	newNode := !utils.PathExists(filepath.Join(store.RaftDir, "raft.db"))
	//
	addr, err := net.ResolveTCPAddr("tcp", store.RaftBind)
	if err != nil {
		return err
	}
	//监听端口
	transport, err := raft.NewTCPTransport(store.RaftBind, addr, 3, 10*time.Second, os.Stderr)
	if err != nil {
		return err
	}
	snapshots, err := raft.NewFileSnapshotStore(store.RaftDir, retainSnapshotCount, os.Stderr)
	if err != nil {
		return fmt.Errorf("file snapshot store: %s", err)

	}

	var logStore raft.LogStore
	var stableStore raft.StableStore

	boltDB, err := raftboltdb.NewBoltStore(filepath.Join(store.RaftDir, "raft.db"))
	if err != nil {
		return fmt.Errorf("new bolt store: %s", err)
	}
	logStore = boltDB
	stableStore = boltDB

	ra, err := raft.NewRaft(config, (*fsm)(store), logStore, stableStore, snapshots, transport)
	if err != nil {
		return fmt.Errorf("new raft: %s", err)
	}
	store.raft = ra

	if enableSingle && newNode {
		store.logger.Printf("bootstrap needed \n")
		configuration := raft.Configuration{
			Servers: []raft.Server{
				{
					ID:      config.LocalID,
					Address: transport.LocalAddr(),
				},
			},
		}
		//启动集群模式
		ra.BootstrapCluster(configuration)
	} else {
		store.logger.Printf("no bootstrap needed")
	}
	return nil
}

// WaitForLeader  阻塞等待领导心跳或者超时
func (store *Store) WaitForLeader(timeout time.Duration) (string, error) {
	tck := time.NewTicker(leaderWaitDelay) //周期性循环
	defer tck.Stop()
	tmr := time.NewTimer(timeout)//延迟触发
	defer tmr.Stop()
	for {
		select {
		case <-tck.C:
			leader, _ := store.LeaderWithID()
			if leader != "" {
				return leader, nil
			}
		case <-tmr.C:
			return "", fmt.Errorf("timeout expired")
		}
	}
}

// WaitForAppliedIndex 阻塞等待日志应用或者超时
func (store *Store) WaitForAppliedIndex(idx uint64, timeout time.Duration) error {
	tck := time.NewTicker(appliedWaitDelay)
	defer tck.Stop()
	tmr := time.NewTimer(timeout)
	defer tmr.Stop()
	for {
		select {
		case <-tck.C:
			if store.raft.AppliedIndex() >= idx {
				return nil
			}
		case <-tmr.C:
			return fmt.Errorf("timeout expired")
		}
	}
}
// FSM 接口实现类
type fsm Store

func (f *fsm) Snapshot() (raft.FSMSnapshot, error) {
	f.lock.Lock()
	defer f.lock.Unlock()
	obj := make(map[string]string)
	for k, v := range f.db {
		obj[k] = v
	}
	return &fsmSnapshot{store: obj}, nil
}
func (f *fsm) Restore(rc io.ReadCloser) error {
	obj := make(map[string]string)
	if err := json.NewDecoder(rc).Decode(&obj); err != nil {
		return err
	}
	f.db = obj
	return nil
}

func (f *fsm) Apply(log *raft.Log) interface{} {
	var c command
	if err := json.Unmarshal(log.Data, &c); err != nil {
		panic(fmt.Sprintf("failed to unmarshal command: %s", err.Error()))
	}
	switch c.Op {
	case "set":
		return f.applySet(c.Key, c.Value)
	case "delete":
		return f.applyDelete(c.Key)
	default:
		panic(fmt.Sprintf("unrecognized command op: %s", c.Op))
	}
}
func (f *fsm) applySet(key, value string) interface{} {
	f.lock.Lock()
	defer f.lock.Unlock()
	f.db[key] = value
	return nil
}
func (f *fsm) applyDelete(key string) interface{} {
	f.lock.Lock()
	defer f.lock.Unlock()
	delete(f.db, key)
	return nil
}

//状态机的快照
type fsmSnapshot struct {
	store map[string]string
}

func (f *fsmSnapshot) Persist(sink raft.SnapshotSink) error {
	err := func() error {
		data, err := json.Marshal(f.store)
		if err != nil {
			return err
		}
		if _, err = sink.Write(data); err != nil {
			return err
		}
		return sink.Close()
	}()
	if err != nil {
		sink.Cancel()
		return err
	}
	return nil
}
func (f *fsmSnapshot) Release() {}
